package com.esdk.utils;

import cn.hutool.core.date.LocalDateTimeUtil;
import com.esdk.esdk;
import com.esdk.interfaces.Callback;
import com.esdk.interfaces.Runner;
import lombok.NonNull;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @author ffychina
 * @since 2024-05-27
 * https://blog.csdn.net/ffychina
 * */
public class ThreadUtils{
	public static int MaximumPoolSize=10000;
	public static int CorePoolSize=100; //如果设为0会容易导致阻塞
	public static int DefaultTimeOutSec=60; //必须>=0，否则无法创建抛nullpointexception
	protected static ThreadPoolExecutor virtualThreadExecutor=createExecutor();


	public static ThreadPoolExecutor createExecutor(){
		//Executors.newVirtualThreadPerTaskExecutor(); //无法限制线程数，弃用
		ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(CorePoolSize,MaximumPoolSize,DefaultTimeOutSec,TimeUnit.SECONDS
			,new ArrayBlockingQueue<>(MaximumPoolSize),Thread.ofVirtual().name("vt-",1).factory());
		return threadPoolExecutor;
	}

	public static ThreadPoolExecutor createExecutor(int corePoolSize,int maxPoolSize,int DefaultTimeOutSec,boolean ifVirtual,String prefixName){
		ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(corePoolSize,maxPoolSize,DefaultTimeOutSec,TimeUnit.SECONDS
			,new ArrayBlockingQueue<>(maxPoolSize*10),ifVirtual?Thread.ofVirtual().name(prefixName,1).factory():Thread.ofPlatform().name(prefixName,1).factory()
//			,new ThreadPoolExecutor.CallerRunsPolicy()
		);
		return threadPoolExecutor;
	}

	public static Thread interval(String threadName,long intervalMs,Runner runner){
		return interval(threadName,intervalMs,runner,null);
	}

	public static Thread interval(String threadName,long intervalMs,Runner runner,Callback<Throwable> catchHandle){
		Thread thread=new Thread(()->{
			while(true){
				try {
					runner.run();
				} catch (Throwable e) {
					esdk.func.IF(catchHandle==null,()->esdk.error(e),()->catchHandle.invoke(e));
				}finally{
					esdk.tool.sleep(intervalMs);
				}
			}
		},threadName);
		thread.start();
		return thread;
	}

	/**
	 * 使用CallerRunsPolicy策略，如果待处理任务数量超出队列数量，在执行submit时会阻塞，并且会调用主线程处理，但不会抛弃超出队伍的任务。
	 * @param corePoolSize 核心线程如果>=1不会自动销毁，需要手工执行awaitTermination(executor)进行销毁
	 * @param DefaultTimeOutSec 连接池实例自动销毁非核心线程的空闲等待时间(秒)，必须>=0，否则无法抛出nullpointexception
	 * */
	public static ThreadPoolExecutor createExecutor(int corePoolSize,int maxPoolSize,int queueCapacity,int DefaultTimeOutSec,boolean ifVirtual,String prefixName){
		ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(corePoolSize,maxPoolSize,DefaultTimeOutSec,TimeUnit.SECONDS
			,new ArrayBlockingQueue<>(queueCapacity),ifVirtual?Thread.ofVirtual().name(prefixName,1).factory():Thread.ofPlatform().name(prefixName,1).factory()
//			,new ThreadPoolExecutor.CallerRunsPolicy()
		);
		return threadPoolExecutor;
	}

	/**等待返回结果*/
	public static <V> V await(Callable<V> callable){
		Future<V> submit=virtualThreadExecutor.submit(callable);
		try{
			V v=submit.get();
			return v;
		}catch(Exception e){
			throw esdk.tool.wrapThrowble(e);
		}
	}

	/**等待返回结果*/
	public static <V> V await(ThreadPoolExecutor executor,Callable<V> callable){
		Future<V> submit=executor.submit(callable);
		try{
			V v=submit.get();
			return v;
		}catch(Exception e){
			throw esdk.tool.wrapThrowble(e);
		}
	}

	/**等待返回结果*/
	public static <V> List<V> await(List<Callable<V>> callables){
		List<Future<V>> futureList=new ArrayList<>();
		for(Callable<V> callable:callables){
			futureList.add(virtualThreadExecutor.submit(callable));
		}
		List<V> resultList=new ArrayList<>();
    try{
			for(Future<V> future:futureList){
				V v=future.get();
				resultList.add(v);
			}
			return resultList;
		}catch(Exception e){
			throw esdk.tool.wrapThrowble(e);
    }
  }


	public static <V> List<V> await(List<Callable<V>> callables,int maxThreadSize){
		return await(callables,maxThreadSize,"pt-");
	}

	/**
	 * 自动创建线程池实例执行批量任务，执行完成后释放该线程池，最大程度并行执行任务，并且在加入队列时会阻塞，避免触发拒绝策略
	 * */
	public static <V> List<V> await(List<Callable<V>> callables,int maxThreadSize,String threadPrefixName){
		int maxPoolSize=esdk.math.min(callables.size(),maxThreadSize);
		ThreadPoolExecutor executor=createExecutor(maxPoolSize,maxPoolSize,maxPoolSize,DefaultTimeOutSec,false,threadPrefixName);
		Semaphore semaphore=new Semaphore(maxThreadSize);
		LinkedList<Future<V>> futureList=new LinkedList<>();
		List<V> resultList=new ArrayList<>();
		try{
			while(!callables.isEmpty()||!futureList.isEmpty()){
				for(int i=0,n=esdk.math.min(callables.size(),semaphore.availablePermits());i<n;i++){
					semaphore.acquire();
					futureList.add(executor.submit(callables.removeFirst()));
				}
				for(int i=0;i<futureList.size();i++){
					Future<V> future=futureList.get(i);
					if(future.isDone()){
						try{
							V v=future.get();
							resultList.add(v);
						}catch(Exception e){
							esdk.error(e);
						}finally{
							futureList.remove(future);
							semaphore.release();
							i--;
						}
					}
				}
				esdk.tool.sleep(10);
			}
			return resultList;
		}catch(Exception e){
			throw esdk.tool.wrapThrowble(e);
		}finally{
			awaitTermination(executor);
		}
	}


	/**
	 * 先把所有任务都加入到线程池，再等待线程处理，如果队列过大有可能会出现oom，或者容易触发拒绝策略。
	 * 注意：不会关闭核心线程，需要外部手工执行awaitTermination(executor);
	 * */
	public static <V> List<V> await(List<Callable<V>> callables,ThreadPoolExecutor executor){
		LinkedList<Future<V>> futureList=new LinkedList<>();
		List<V> resultList=new ArrayList<>();
		for(Callable<V> callable: callables){
			futureList.add(executor.submit(callable));
		}
		while(!futureList.isEmpty()){
			for(int i=0;i<futureList.size();i++){
				Future<V> future=futureList.get(i);
				if(future.isDone()){
					try{
						V v=future.get();
						resultList.add(v);
					}catch(Exception e){
						esdk.error(e);
					}finally{
						futureList.remove(future);
						i--;
					}
				}
			}
			esdk.tool.sleep(10);
		}
		return resultList;
	}


	public static void async(ThreadPoolExecutor executor,Runner runner){
		executor.execute(()->{
			try{
				runner.run();
			}catch(Exception e){
				throw esdk.tool.wrapThrowble(e);
			}
		});
	}


	public static <R,V> void async(ThreadPoolExecutor executor,Callable<V> callable,Callback<V> callback){
		Future<V> future=executor.submit(callable);
    try{
			V v=future.get();
      callback.call(v);
    }catch(Exception e){
      throw esdk.tool.wrapThrowble(e);
    }
  }

	/**返回结果给回调函数*/
	public static <R,V> void async(Callable<V> callable,Callback<V> callback){
		virtualThreadExecutor.execute(()->{
			try{
				V v=callable.call();
				callback.invoke(v);
			}catch(Exception e){
				throw esdk.tool.wrapThrowble(e);
			}
		});
	}

	/**不返回结果*/
	public static void async(Runner runner){
		virtualThreadExecutor.execute(()->{
      try{
        runner.run();
      }catch(Exception e){
       	throw esdk.tool.wrapThrowble(e);
      }
    });
	}

	/**
	 * 不返回结果
	 */
	public static void async(List<Runner> runners){
		CountDownLatch countDownLatch=new CountDownLatch(runners.size());
		for(Runner runner: runners){
			virtualThreadExecutor.execute(()->{
				try{
					runner.run();
					countDownLatch.countDown();
				}catch(Exception e){
					throw esdk.tool.wrapThrowble(e);
				}
			});
		}
		try{
			countDownLatch.await();
		}catch(InterruptedException e){
			throw esdk.tool.wrapThrowble(e);
		}
	}

	public static void awaitTermination(ThreadPoolExecutor executor){
		awaitTermination(executor,DefaultTimeOutSec,TimeUnit.SECONDS);
	}

	public static void awaitTermination(ThreadPoolExecutor executor,long timeout,TimeUnit timeUnit){
		awaitTermination(executor,timeout,timeUnit,true);
	}

	public static void awaitTermination(){
		awaitTermination(DefaultTimeOutSec,TimeUnit.SECONDS);
	}

	public static void awaitTermination(long timeout,TimeUnit timeUnit){
		awaitTermination(timeout,timeUnit,true);
	}

	public static void awaitTermination(long timeout,TimeUnit timeUnit,boolean isForceShutdown){
		awaitTermination(virtualThreadExecutor,timeout,timeUnit,isForceShutdown);
	}

	/** 等待所有线程处理结束
	 * 注意：isShutdown()表明关闭过程已经开始，而isTerminated()则表明关闭过程已经完成
	 * */
	public static void awaitTermination(@NonNull ThreadPoolExecutor executor,long timeout,TimeUnit timeUnit,boolean isForceShutdown){
		synchronized(executor){
			try{
				if(isForceShutdown){
					executor.shutdown();// 通知线程池不再接收新任务
					boolean result=executor.awaitTermination(timeout,timeUnit); //等60秒
					if(!result){
						List<Runnable> runnables=executor.shutdownNow();// 强制关闭，返回尚未开始执行的线程列表
						int activeCount=executor.getActiveCount();
						esdk.warn("有{}个活动线程未执行完成，其中{}个仍没开始执行，已强制终止！",activeCount,runnables.size());
					}
					if(executor.equals(virtualThreadExecutor))
						virtualThreadExecutor=createExecutor();
				}else{
					TimeMeter tm=new TimeMeter();
					TimeMeter tm1=new TimeMeter();
					while(executor.getActiveCount()>0){
						if(tm1.getElapse()>timeUnit.toMillis(timeout)){
							esdk.warn("有{}个活动线程已超时{}仍未执行完成，继续等待中...",executor.getActiveCount(),tm.getHumanElapse());
							tm1.start();
						}
						esdk.tool.sleep(100);
					}
					esdk.sout("所有线程已全部执行完成");
				}
			}catch(InterruptedException e){
				throw esdk.tool.wrapThrowble(e);
			}
		}
	}

	public static void shutdownOnCompleted(@NonNull ThreadPoolExecutor executor) {
		shutdownOnCompleted(executor,DefaultTimeOutSec);
	}

	public static void shutdownOnCompleted(@NonNull ThreadPoolExecutor executor,long waitSleepSec) {
		executor.shutdown();
		while(executor.getActiveCount()>0) {
			try{
				if (!executor.awaitTermination(waitSleepSec, TimeUnit.SECONDS)) {
					// 如果等待时间超过了60秒，输出提示信息
					esdk.swarn("所有线程在等待完成时关闭过程中出现异常，线程池未能在"+waitSleepSec+"秒内完成所有任务的执行");
				}
			}catch(InterruptedException e){
				executor.shutdownNow();
				throw new RuntimeException(e);
			}
		}
		executor.shutdownNow();
	}

	/**注意：isShutdown()表明关闭过程已经开始，而isTerminated()则表明关闭过程已经完成*/
	public static boolean isShutdown(){
		return virtualThreadExecutor.isShutdown();
	}

	/**注意：isShutdown()表明关闭过程已经开始，而isTerminated()则表明关闭过程已经完成*/
	public static boolean isTerminated(){
		return virtualThreadExecutor.isTerminated();
	}

	public static void close(){
		virtualThreadExecutor.close();
	}

	public static Map getExecutorsInfo(){
		return getExecutorsInfo(virtualThreadExecutor);
	}

	public static Map getExecutorsInfo(ThreadPoolExecutor executor){
		long threadCount=(long)esdk.reflect.getDeclaredMethodValue(executor,"threadCount");
//		Stream<Thread> threadStream=(Stream<Thread>)esdk.reflect.getDeclaredMethodValue(executor,"threads");
		int aliveSize=0,interruptedSize=0;
		return esdk.map.createMap("threadCount",threadCount,"aliveSize",aliveSize,"interruptedSize",interruptedSize);
	}

	public static void printExecutorInfo(ThreadPoolExecutor executor){
			System.out.println("=== 线程池状态监控 ===");
			System.out.println("核心线程数: " + executor.getCorePoolSize());
			System.out.println("最大线程数: " + executor.getMaximumPoolSize());
			System.out.println("当前线程数: " + executor.getPoolSize());
			System.out.println("活跃线程数: " + executor.getActiveCount());
			System.out.println("已完成任务数: " + executor.getCompletedTaskCount());
			System.out.println("总任务数: " + executor.getTaskCount());
			System.out.println("队列中等待的任务数: " + executor.getQueue().size());
			System.out.println("========================");
	}

	private static void test1(){
		esdk.sout("开始计时");
		TimeMeter tm1=TimeMeter.startNew();
		ThreadUtils.async(()->{
			esdk.tool.sleep(100);
			esdk.sout("运行结束");
		});
		ThreadUtils.async(()->{
			esdk.tool.sleep(500);
			return LocalDateTimeUtil.now();
		},(now)->{
			esdk.sout("异步返回值:{},耗时:{}",now,tm1.getHumanElapse());
		});
//		esdk.sout(ThreadUtils.getExecutorsInfo());
		TimeMeter tm2=TimeMeter.startNew();
		Object v=ThreadUtils.await(()->{
			esdk.tool.sleep(200);
			return LocalDateTimeUtil.now();
		});
		esdk.sout("同步返回值:{},耗时:{}",v,tm2.getHumanElapse());
//		executorService.awaitTermination(50,TimeUnit.MILLISECONDS);
		ThreadUtils.close();
		ThreadUtils.awaitTermination();
		esdk.sout("搞完收工，耗时{}",tm1.getHumanElapse());
	}

	static void test2(){
		ArrayList<Callable<String>> tasks=new ArrayList<>();
		for(int i=0;i<20;i++){
			int finalI=i+1;
			tasks.add(()->{
				String threadName=Thread.currentThread().getName();
				String taskName="任务"+finalI;
				esdk.sout("{}开始：{}，时间{}:",taskName,threadName,LocalDateTime.now());
				esdk.tool.sleep(5000);
				esdk.sout("{}完成：{}，时间{}:",taskName,threadName,LocalDateTime.now());
				return taskName;
			});
		}
		List<String> ress=ThreadUtils.await(tasks,4,"测试");
		esdk.sout(esdk.str.valueOf(ress));
		esdk.tool.printUpTime();
	}

	public static void main(String[] args) throws InterruptedException{
//		test1();
		test2();
	}
}

